package io.debezium.connector.postgresql.connection;

import io.debezium.connector.postgresql.TypeRegistry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.ByteBuffer;
import java.sql.SQLException;

public abstract class AbstractMessageDecoder implements MessageDecoder {
    private static final Logger LOGGER = LoggerFactory.getLogger(AbstractMessageDecoder.class);

    public AbstractMessageDecoder() {}

    public void processMessage(
            ByteBuffer buffer,
            ReplicationStream.ReplicationMessageProcessor processor,
            TypeRegistry typeRegistry)
            throws SQLException, InterruptedException {
        if (buffer == null) {
            processor.process(null);
        } else {
            this.processNotEmptyMessage(buffer, processor, typeRegistry);
        }
    }

    protected abstract void processNotEmptyMessage(
            ByteBuffer var1, ReplicationStream.ReplicationMessageProcessor var2, TypeRegistry var3)
            throws SQLException, InterruptedException;

    public boolean shouldMessageBeSkipped(
            ByteBuffer buffer, Lsn lastReceivedLsn, Lsn startLsn, WalPositionLocator walPosition) {
        if (walPosition.skipMessage(lastReceivedLsn)) {
            LOGGER.info(
                    "Streaming requested from LSN {}, received LSN {} identified as already processed",
                    startLsn,
                    lastReceivedLsn);
            return true;
        } else {
            return false;
        }
    }

    public void close() {}
}
